package org.project.join

import org.FlinkStreamApp
import org.apache.flink.api.common.state.ValueStateDescriptor
import org.apache.flink.api.scala.typeutils.Types
import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

/**
 * description ：双流 join
 * author      ：剧情再美终是戏 
 * mail        : 13286520398@163.com
 * date        ：Created in 2020/2/27 14:51
 * modified By ：
 * version:    : 1.0
 */

case class OrderEvent(orderId: String,
                      eventType: String,
                      eventTime: String)

case class PayEvent(orderId: String,
                    eventType: String,
                    eventTime: String)

object TwoStreamsJoin extends FlinkStreamApp {

  // 创建两个侧输出流标签
  val orderOutput = new OutputTag[OrderEvent]("orderOutput")
  val payOutput = new OutputTag[PayEvent]("payOutput")

  override def doSomeThing(environment: StreamExecutionEnvironment) = {
    // 创建订单数据和支付信息
    val orders = environment
      .fromCollection(List(
        OrderEvent("1", "create", "1558430842"),
        OrderEvent("2", "create", "1558430843"),
        OrderEvent("1", "pay", "1558430844"),
        OrderEvent("2", "pay", "1558430845"),
        OrderEvent("3", "create", "1558430849"),
        OrderEvent("3", "pay", "1558430849")
      ))
      .assignAscendingTimestamps(_.eventTime.toLong * 1000)
      .keyBy("orderId")

    val pays = environment.fromCollection(List(
      PayEvent("1", "weixin", "1558430847"),
      PayEvent("2", "zhifubao", "1558430848"),
      PayEvent("4", "zhifubao", "1558430850")
    )).assignAscendingTimestamps(_.eventTime.toLong * 1000)
      .keyBy("orderId")


    // 双流 join

    val result = orders
      .connect(pays)
      .process(new OrderPayProcess())

    // 正常匹配的订单和支付信息如下
    result.print()

    // 未匹配到支付信息的订单如下
    result.getSideOutput(orderOutput).print()

    // 未匹配到订单的地支付信息如下
    result.getSideOutput(payOutput).print()
  }

  class OrderPayProcess extends KeyedCoProcessFunction[String, OrderEvent, PayEvent, (OrderEvent, PayEvent)] {
    // 创建两个状态来保存 order 和 pay 迟到的状态
    lazy val orderState = getRuntimeContext.getState(
      new ValueStateDescriptor[OrderEvent]("orderState", Types.of[OrderEvent])
    )
    lazy val payState = getRuntimeContext.getState(
      new ValueStateDescriptor[PayEvent]("payState", Types.of[PayEvent])
    )

    // 处理订单流
    override def processElement1(order: OrderEvent, ctx: KeyedCoProcessFunction[String, OrderEvent, PayEvent, (OrderEvent, PayEvent)]#Context, out: Collector[(OrderEvent, PayEvent)]) = {
      // 如果订单流来了数据，就去看一下支付状态中有没有数据
      if (null != payState.value()) {
        out.collect(order, payState.value())
        payState.clear()
      } else { // 如果支付状态中没有数据，就当当前订单数据保存起来
        orderState.update(order)
        // 注册一个定时器，设置为等待 1s
        ctx.timerService().registerEventTimeTimer(order.eventTime.toLong * 1000)
      }
    }

    // 处理支付流
    override def processElement2(pay: PayEvent, ctx: KeyedCoProcessFunction[String, OrderEvent, PayEvent, (OrderEvent, PayEvent)]#Context, out: Collector[(OrderEvent, PayEvent)]) = {
      // 如果支付流来了数据，就去看一下订单流状态是否有数据
      if (null != orderState.value()) {
        out.collect(orderState.value(), pay)
        orderState.clear()
      } else { // 如果订单状态中没有数据，就将当前支付信息保存起来
        payState.update(pay)
        // 注册一个定时器，设置为等待 1s
        ctx.timerService().registerEventTimeTimer(pay.eventTime.toLong * 1000)
      }
    }

    // 触发定时器
    override def onTimer(timestamp: Long, ctx: KeyedCoProcessFunction[String, OrderEvent, PayEvent, (OrderEvent, PayEvent)]#OnTimerContext, out: Collector[(OrderEvent, PayEvent)]) = {
      // 如果订单状态不为空，就将其输出到侧输出流中
      if (null != orderState.value()) {
        ctx.output(orderOutput, orderState.value())
        orderState.clear()
      }

      // 如果支付状态不为空，就将其输出到侧输出流中
      if (null != payState.value()) {
        ctx.output(payOutput, payState.value())
        payState.clear()
      }
    }
  }

}
